package com.dji.sample.component.websocket.service.impl;

import com.dji.sample.common.constant.CommonConstant;
import com.dji.sample.common.model.SocketMsg;
import com.dji.sample.component.mqtt.config.MqttConfiguration;
import com.dji.sample.component.mqtt.model.ChannelName;
import com.dji.sample.component.mqtt.service.IMqttMessageGateway;
import com.dji.sample.component.websocket.config.ConcurrentWebSocketSession;
import com.dji.sample.component.websocket.model.CustomWebSocketMessage;
import com.dji.sample.component.websocket.service.ISendMessageService;
import com.dji.sample.component.websocket.service.IWebSocketManageService;
import com.dji.sample.manage.model.receiver.FirmwareVersionReceiver;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import org.springframework.web.socket.TextMessage;

import java.io.IOException;
import java.util.Collection;

import static com.dji.sample.common.constant.CommonConstant.DEFAULT_WORK_SPACE_ID;

/**
 * @author sean.zhou
 * @version 0.1
 * @date 2021/11/24
 */
@Service
@Slf4j
public class SendMessageServiceImpl implements ISendMessageService {

    @Value("${node.mode}")
    private String mode;

    @Autowired
    MqttConfiguration mqttConfiguration;

    @Autowired
    IMqttMessageGateway mqttGateway;

    @Autowired
    private IWebSocketManageService webSocketManageService;

    @Autowired
    private ObjectMapper mapper;

    @Override
    public void sendMessage(ConcurrentWebSocketSession session, CustomWebSocketMessage message) {
        if (session == null) {
            return;
        }
        try {
            if (!session.isOpen()) {
                session.close();
                log.debug("This session is closed.");
                return;
            }
            session.sendMessage(new TextMessage(mapper.writeValueAsBytes(message)));
        } catch (IOException e) {
            log.info("Failed to publish the message. {}", message.toString());
            e.printStackTrace();
        }
    }

    @Override
    public void sendMessage(ConcurrentWebSocketSession session, CustomWebSocketMessage message, String userId, String sessionId) {
        //单发 当前session 为空 可能在其它节点 广播消息
        if (session == null) {
            if (!mode.equals(CommonConstant.STAND_ALONE)) {
                SocketMsg socketMsg = new SocketMsg();
                try {
                    socketMsg.setReceiveId(userId);
                    socketMsg.setSessionId(sessionId);
                    socketMsg.setMsg(mapper.writeValueAsBytes(message));
                    socketMsg.setType(Integer.MIN_VALUE);
                    socketMsg.setClientId(mqttConfiguration.getClientId());
                    mqttGateway.publish(CommonConstant.SOCKET_TOPIC, mapper.writeValueAsBytes(socketMsg));
                } catch (JsonProcessingException e) {
                    log.error("JsonProcessingException", e);
                }
            }
            return;
        }
        try {
            if (!session.isOpen()) {
                session.close();
                log.debug("This session is closed.");
                return;
            }
            session.sendMessage(new TextMessage(mapper.writeValueAsBytes(message)));
        } catch (IOException e) {
            log.info("Failed to publish the message. {}", message.toString());
            e.printStackTrace();
        }
    }

    // 集群模式群发消息各节点监听到之后判断是有满足同条件的session 有则推送消息没有则不处理
    @Override
    public void sendBatch(Integer userType, Collection<ConcurrentWebSocketSession> sessions, CustomWebSocketMessage message) {
        try {
            //集群模式 消息推送到其它节点
            if (!mode.equals(CommonConstant.STAND_ALONE)) {
                SocketMsg socketMsg = new SocketMsg();
                socketMsg.setMsg(mapper.writeValueAsBytes(message));
                socketMsg.setType(userType);
                socketMsg.setClientId(mqttConfiguration.getClientId());
                mqttGateway.publish(CommonConstant.SOCKET_TOPIC, mapper.writeValueAsBytes(socketMsg));
            }
            if (sessions.isEmpty()) {
                return;
            }
            TextMessage data = new TextMessage(mapper.writeValueAsBytes(message));

            for (ConcurrentWebSocketSession session : sessions) {
                if (!session.isOpen()) {
                    session.close();
                    log.debug("This session is closed.");
                    return;
                }
                session.sendMessage(data);
            }

        } catch (IOException e) {
            log.error("Failed to publish the message", e);
        }
    }

    @ServiceActivator(inputChannel = ChannelName.SOCKET)
    public void socketMsgListener(Message<?> message) {
        try {
            SocketMsg socketMsg = mapper.readValue((byte[]) message.getPayload(), SocketMsg.class);
            //当前节点发出的消息 不再处理
            if (socketMsg.getClientId().equals(mqttConfiguration.getClientId())) {
                return;
            }
            Collection<ConcurrentWebSocketSession> sessions = null;
            if (socketMsg.getType() == Integer.MAX_VALUE) {
                //群发
                sessions = webSocketManageService.getValueWithWorkspace(DEFAULT_WORK_SPACE_ID);
            } else if (socketMsg.getType() == Integer.MIN_VALUE && StringUtils.hasText(socketMsg.getSessionId())) {
                //单发
                ConcurrentWebSocketSession session = WebSocketManageServiceImpl.SESSIONS.get(socketMsg.getSessionId());
                if (session == null) {
                    return;
                }
                if (!session.isOpen()) {
                    session.close();
                    return;
                }
                session.sendMessage(new TextMessage(socketMsg.getMsg()));
                return;
            } else {
                //带条件群发
                sessions = webSocketManageService.getValueWithWorkspaceAndUserType(DEFAULT_WORK_SPACE_ID, socketMsg.getType());
            }
            if (sessions.isEmpty()) {
                return;
            }
            TextMessage data = new TextMessage(socketMsg.getMsg());
            for (ConcurrentWebSocketSession session : sessions) {
                if (!session.isOpen()) {
                    session.close();
                    log.debug("This session is closed.");
                    return;
                }
                session.sendMessage(data);
            }

        } catch (IOException e) {
            log.error("socket消息转化异常", e);
        }
    }
}